use std::cmp::Ordering;

use crate::options::Options;
use crate::types::LdbIterator;

use bytes::{Bytes, BytesMut};

/// 在 Rust 中，integer_encoding::FixedInt 和 integer_encoding::VarInt 是 integer-encoding crate 提供的两个 trait，用于整数的编码和解码操作，
/// 主要用于将整数转换为字节流（序列化）或从字节流还原为整数（反序列化）。
///
/// FixedInt trait：提供固定长度的整数编码 / 解码功能。
///
/// 对于相同类型的整数（如 u32），编码后始终占用固定字节数（如 u32 始终占 4 字节）。
///
/// 常用方法：encode_fixed（编码为字节）、decode_fixed（从字节解码）。
///
/// 适合需要固定空间、快速编码解码的场景。
///
/// VarInt trait：提供可变长度的整数编码 / 解码功能（可变整数编码）。
///
/// 编码后的字节长度根据整数的实际大小动态变化（小整数占用更少字节）。
///
/// 常用方法：encode_var（编码为可变长度字节）、decode_var（从字节解码）。
///
/// 适合节省空间的场景，常见于网络传输、数据存储等（如 Protocol Buffers 中的变长编码）。
///
/// ```
/// use integer_encoding::{FixedInt, VarInt};
///
/// fn main() {
///     // 固定长度编码
///     let num: u32 = 42;
///     let mut buf = [0; 4];
///     num.encode_fixed(&mut buf);
///     let decoded = u32::decode_fixed(&buf).unwrap();
///
///     // 可变长度编码
///     let num: u64 = 12345;
///     let mut var_buf = Vec::new();
///     num.encode_var(&mut var_buf);
///     let (decoded_var, _) = u64::decode_var(&var_buf).unwrap();
/// }
/// ```
use integer_encoding::FixedInt;
use integer_encoding::VarInt;

/// 这段 Rust 代码定义了一个名为 BlockContents 的公共类型别名（type alias），它实际上是 Bytes 类型的别名。
///
/// 具体来说：
///
/// pub 表示这个类型别名是公共的，可以被其他模块访问
///
/// type BlockContents 声明了一个新的类型别名
///
/// = Bytes 表示这个别名指向 Bytes 类型
///
/// Bytes 通常是指 bytes::Bytes 类型（来自 bytes crate），这是一个高效的字节容器类型，常用于网络编程、IO 操作等场景，具有以下特点：
///
/// 可以高效地共享和克隆字节数据（通过引用计数）
///
/// 支持零成本切片操作
///
/// 比普通的 Vec<u8> 更适合用于需要频繁传递和处理字节数据的场景
///
/// 所以这段代码的作用是：为 Bytes 类型创建一个更具业务含义的别名 BlockContents，让代码更具可读性，明确表示这个类型用于存储 "区块内容"（从命名推测）的字节数据。
///
/// 例如，后续代码中可以直接使用 BlockContents 来声明变量，而不需要每次都写 Bytes，使代码意图更清晰：
///
/// let contents: BlockContents = Bytes::from("some block data");
pub type BlockContents = Bytes;

/// A Block is an immutable ordered set of key/value entries.
///
/// The structure internally looks like follows:
///
/// A block is a list of ENTRIES followed by a list of RESTARTS, terminated by a fixed u32
/// N_RESTARTS.
///
/// An ENTRY consists of three varints, SHARED, NON_SHARED, VALSIZE, a KEY and a VALUE.
///
/// SHARED denotes how many bytes the entry's key shares with the previous one.
///
/// NON_SHARED is the size of the key minus SHARED.
///
/// VALSIZE is the size of the value.
///
/// KEY and VALUE are byte strings; the length of KEY is NON_SHARED.
///
/// A RESTART is a fixed u32 pointing to the beginning of an ENTRY.
///
/// N_RESTARTS contains the number of restarts.
///
/// 这段代码定义了一个 Rust 结构体 Block，用于表示一个不可变的键值对条目有序集合，通常出现在类似 LevelDB 或 RocksDB 的存储引擎实现中。
///
/// 具体解析如下：
///
/// 文档注释说明：
///
/// 明确 Block 是一个不可变的键值对有序集合
///
/// 详细描述了内部结构：
///
/// 由 ENTRIES（条目列表）、RESTARTS（重启点列表）和 N_RESTARTS（重启点数量，u32 类型）组成
///
/// 每个 ENTRY 包含三个 varint 类型（可变长度整数）：SHARED（与前一键共享的字节数）、NON_SHARED（键的非共享部分长度）、VALSIZE（值的长度），以及实际的 KEY 和 VALUE 字节数据
///
/// RESTART 是指向某个 ENTRY 起始位置的 u32 偏移量
///
/// N_RESTARTS 存储重启点的数量
///
/// 结构体定义：
///
/// #[derive(Clone)] 派生了 Clone 特性，允许该结构体被克隆
///
/// pub struct Block 定义了一个公共结构体
///
/// 包含两个字段：
///
/// block: BlockContents：存储区块的实际字节数据（使用之前定义的 BlockContents 类型，即 Bytes 的别名）
///
/// opt: Options：可能是区块的配置选项（具体内容需看 Options 结构体的定义）
///
/// 这个 Block 结构体很可能用于高效存储和检索键值对数据，其中的 "重启点"（restarts）设计是为了优化查找性能 —— 当查找某个键时，可以先通过重启点快速定位到可能包含目标键的条目范围，而不必遍历所有条目。
///
/// 这种结构在高性能键值存储中很常见，通过巧妙的内存布局设计，在空间效率和查询性能之间取得平衡。
#[derive(Clone)]
pub struct Block {
    block: BlockContents,
    opt: Options,
}

impl Block {
    /// Return an iterator over this block.
    /// Note that the iterator isn't bound to the block's lifetime; the iterator uses the same
    /// refcounted block contents as this block, meaning that if the iterator isn't released,
    /// the memory occupied by the block isn't, either)
    pub fn iter(&self) -> BlockIter {
        let restarts = u32::decode_fixed(&self.block[self.block.len() - 4..]);
        let restart_offset = self.block.len() - 4 - 4 * restarts as usize;

        BlockIter {
            block: self.block.clone(),
            opt: self.opt.clone(),

            offset: 0,
            restarts_off: restart_offset,
            current_entry_offset: 0,
            current_restart_ix: 0,

            key: BytesMut::new(),
            val_offset: 0,
        }
    }

    pub fn contents(&self) -> BlockContents {
        self.block.clone()
    }

    pub fn new(opt: Options, contents: BlockContents) -> Block {
        assert!(contents.len() > 4);
        Block {
            block: contents,
            opt,
        }
    }
}

/// BlockIter is an iterator over the entries in a block. It doesn't depend on the Block's
/// lifetime, as it uses a refcounted block underneath.
///
/// 这段代码定义了一个名为 BlockIter 的结构体，它是 SSTable 数据块（Block）的迭代器，用于遍历数据块中的键值对条目。
/// 它的核心作用是提供对 SSTable 数据块的高效访问接口，屏蔽底层存储细节（如前缀压缩、重启点等），让使用者可以方便地遍历、查找数据块中的键值对。
pub struct BlockIter {
    /// The underlying block contents.
    /// TODO: Maybe (probably...) this needs an Arc.
    /// 存储底层数据块的原始内容（二进制数据），是迭代器访问数据的基础。注释中提到可能需要用 Arc（原子引用计数）来管理，推测是为了支持多线程安全共享数据块。
    block: BlockContents,
    /// 包含数据库的配置选项（如键比较器 cmp），用于指导迭代器的行为（如键的排序规则）。
    opt: Options,
    /// offset of restarts area within the block.
    /// 记录数据块中重启点数组的起始偏移量（重启点是优化查询的关键结构，存储在数据块末尾）。
    restarts_off: usize,

    /// start of next entry to be parsed.
    /// 下一个要解析的条目的起始偏移量（控制迭代进度）。
    offset: usize,
    /// offset of the current entry.
    /// 当前条目的起始偏移量（标记当前访问的条目位置）。
    current_entry_offset: usize,
    /// index of the most recent restart.
    /// 当前所在的重启点索引（用于快速定位和优化反向遍历）。
    current_restart_ix: usize,

    /// We assemble the key from two parts usually, so we keep the current full key here.
    /// 缓存当前条目的完整键（通过 assemble_key 方法组装，包含共享前缀和非共享部分）。
    key: BytesMut,
    /// Offset of the current value within the block.
    /// 当前条目的值（Value）在数据块中的起始偏移量（用于快速提取值）。
    val_offset: usize,
}

impl BlockIter {
    /// Return the number of restarts in this block.
    ///
    /// 这段代码定义了 number_restarts 方法，用于获取 SSTable 数据块（Block）中重启点（restart points）的数量。
    fn number_restarts(&self) -> usize {
        // SSTable 的数据块在存储时，会在末尾用 4 个字节存储重启点的数量（以无符号 32 位整数 u32 格式编码）
        // 方法通过 self.block.len() - 4.. 截取数据块的最后 4 个字节（这是存储重启点数量的固定位置）
        // 使用 u32::decode_fixed 从这 4 个字节中解码出整数（这是一种固定大小的整数解码方式）
        // 最后将其转换为 usize 类型返回，代表当前数据块中重启点的总数
        u32::decode_fixed(&self.block[self.block.len() - 4..]) as usize
    }

    /// Seek to restart point `ix`. After the seek, current() will return the entry at that restart point.
    ///
    /// 这段代码定义了 seek_to_restart_point 方法，用于将迭代器定位到 SSTable 数据块中指定索引（ix）的重启点（restart point）对应的条目，是基于重启点进行高效查找的核心操作。
    fn seek_to_restart_point(&mut self, ix: usize) -> Option<()> {
        // 通过 self.get_restart_point(ix) 获取第 ix 个重启点在数据块中的实际偏移量（即该重启点指向的数据位置）
        let off = self.get_restart_point(ix);
        // 将迭代器的当前偏移量（self.offset）、当前条目偏移量（self.current_entry_offset）设置为重启点的偏移量
        self.offset = off;
        self.current_entry_offset = off;
        // 记录当前重启点的索引（self.current_restart_ix = ix），用于后续跟踪
        self.current_restart_ix = ix;
        // advances self.offset to point to the next entry
        // 调用 self.parse_entry_and_advance() 解析重启点位置的条目数据（包括共享前缀长度、非共享部分长度等元信息），并推进偏移量到下一个条目
        let (shared, non_shared, _, head_len) = self.parse_entry_and_advance()?;
        // 断言 shared == 0（重启点的键没有共享前缀，是完整存储的，这是重启点的特性）
        assert_eq!(shared, 0);
        // 通过 self.assemble_key 基于解析出的信息组装完整的键（由于 shared = 0，直接使用非共享部分作为完整键）
        self.assemble_key(off + head_len, shared, non_shared);
        // 验证迭代器状态有效（assert!(self.valid())），确保定位成功
        assert!(self.valid());
        Some(())
    }

    /// Return the offset that restart `ix` points to.
    ///
    /// 这段代码定义了 get_restart_point 方法，用于获取 SSTable 数据块（Block）中指定索引（ix）的重启点（restart point）所对应的偏移量。
    fn get_restart_point(&self, ix: usize) -> usize {
        // 计算重启点在数据块中的存储位置：
        // self.restarts_off 表示重启点数组在数据块中的起始偏移量（即所有重启点信息的存储起点）
        // 每个重启点的偏移量用 4 个字节（u32）存储，因此第 ix 个重启点的存储位置是 self.restarts_off + 4 * ix
        let restart = self.restarts_off + 4 * ix;
        // 解码偏移量值：
        // 从计算出的位置截取 4 个字节（&self.block[restart..restart + 4]）
        // 使用 u32::decode_fixed 解码这 4 个字节，得到该重启点对应的实际数据偏移量（即这个重启点在数据块中指向的具体位置）
        // 转换为 usize 类型返回
        u32::decode_fixed(&self.block[restart..restart + 4]) as usize
    }

    /// The layout of an entry is
    /// [SHARED varint, NON_SHARED varint, VALSIZE varint, KEY (NON_SHARED bytes),
    ///  VALUE (VALSIZE bytes)].
    ///
    /// Returns SHARED, NON_SHARED, VALSIZE and [length of length spec] from the current position,
    /// where 'length spec' is the length of the three values in the entry header, as described
    /// above.
    /// Advances self.offset to the beginning of the next entry.
    ///
    /// 这段代码定义了 parse_entry_and_advance 方法，用于解析 SSTable 数据块（Block）中单个键值对条目的格式，并推进迭代器到下一个条目的位置。它是读取 SSTable 数据时解析内部存储格式的核心逻辑。
    ///
    /// SSTable 中的键值对条目采用紧凑格式存储，结构为：
    ///
    /// [共享前缀长度(SHARED, varint)] + [非共享前缀长度(NON_SHARED, varint)] + [值长度(VALSIZE, varint)] + [非共享键部分(KEY)] + [值(VALUE)]
    fn parse_entry_and_advance(&mut self) -> Option<(usize, usize, usize, usize)> {
        // 从当前偏移量（self.offset）开始，依次解析：
        // shared：当前键与前一个键的共享前缀长度（通过 usize::decode_var 解码变长整数）
        // non_shared：当前键独有的非共享部分长度
        // valsize：值（VALUE）的长度
        // 同时记录每个变长整数的字节数（sharedlen、non_sharedlen、valsizelen），用于计算总偏移量
        let mut i = 0;
        let (shared, sharedlen) = usize::decode_var(&self.block[self.offset..])?;
        i += sharedlen;
        let (non_shared, non_sharedlen) = usize::decode_var(&self.block[self.offset + i..])?;
        i += non_sharedlen;
        let (valsize, valsizelen) = usize::decode_var(&self.block[self.offset + i..])?;
        i += valsizelen;
        // self.val_offset：记录值（VALUE）在数据块中的起始位置（当前偏移量 + 所有元信息长度 + 非共享键长度）
        // self.offset：更新为下一个条目的起始位置（当前值的结束位置 = self.val_offset + valsize
        self.val_offset = self.offset + i + non_shared;
        self.offset = self.val_offset + valsize;
        // 返回 (shared, non_shared, valsize, i)，其中 i 是三个元信息（SHARED、NON_SHARED、VALSIZE）的总字节数，用于后续组装完整键。
        Some((shared, non_shared, valsize, i))
    }

    /// Assemble the current key from shared and non-shared parts (an entry usually contains only
    /// the part of the key that is different from the previous key).
    ///
    /// `off` is the offset of the key string within the whole block (self.current_entry_offset
    /// + entry header length); `shared` and `non_shared` are the lengths of the shared
    /// respectively non-shared parts of the key.
    /// Only self.key is mutated.
    ///
    /// 这段代码定义了 assemble_key 方法，用于从共享前缀和非共享部分组装出完整的键（key），是处理 SSTable 中键前缀压缩存储的核心逻辑。
    ///
    /// SSTable 为了节省存储空间，对连续的键采用前缀压缩存储 —— 只存储当前键与前一个键的不同部分（非共享部分），而相同部分（共享前缀）通过长度记录（shared）复用前键的内容。
    /// 这个方法的作用就是将这种压缩格式还原为完整的键。
    ///
    /// 方法只修改 self.key（当前键的缓存），不改变其他状态，确保组装结果仅用于当前条目的访问。
    fn assemble_key(&mut self, off: usize, shared: usize, non_shared: usize) {
        // self.key.truncate(shared)：保留当前键中与前一个键共享的前缀部分（截断到 shared 长度）。例如，若前键是 "apple"，当前键的 shared 为 3，则先保留 "app"。
        self.key.truncate(shared);
        // 若存在非共享部分（non_shared > 0）：
        if non_shared > 0 {
            // 从数据块的指定位置（off 到 off + non_shared）截取当前键的独有序列（非共享部分）
            let block_slice_ref: &[u8] = self.block.as_ref();
            // 通过 extend_from_slice 将这部分追加到共享前缀后，形成完整的键。例如，非共享部分是 "ricot"，则完整键为 "app" + "ricot" = "appricot"
            self.key.extend_from_slice(&block_slice_ref[off..off + non_shared]);
        }
    }

    pub fn seek_to_last(&mut self) -> Option<()> {
        if self.number_restarts() > 0 {
            let num_restarts = self.number_restarts();
            self.seek_to_restart_point(num_restarts - 1)?;
        } else {
            self.reset();
        }

        // Stop at last entry, before the iterator becomes invalid.
        //
        // We're checking the position before calling advance; if a restart point points to the
        // last entry, calling advance() will directly reset the iterator.
        while self.offset < self.restarts_off {
            self.advance();
        }
        assert!(self.valid());
        Some(())
    }
}

impl LdbIterator for BlockIter {
    /// 这段代码定义了 advance 方法，用于将 SSTable 数据块（Block）的迭代器向前推进到下一个键值对条目，是迭代器遍历数据的核心逻辑。
    ///
    /// 正向迭代（advance()）
    fn advance(&mut self) -> bool {
        println!("----------block advance----------");
        // 判断是否已到数据块末尾：
        // 若当前偏移量（self.offset）已超过重启点数组的起始位置（self.restarts_off），说明已遍历完所有有效数据条目。
        // 此时调用 self.reset() 重置迭代器状态，并返回 false 表示无法继续推进。
        // 否则，记录当前条目偏移量（self.current_entry_offset = self.offset），准备解析下一个条目。
        if self.offset >= self.restarts_off {
            self.reset();
            return false;
        } else {
            self.current_entry_offset = self.offset;
        }

        let current_off = self.current_entry_offset;
        // 解析下一个条目并组装键：
        // 调用 self.parse_entry_and_advance() 解析下一个条目的元信息（共享前缀长度、非共享部分长度等），并自动推进偏移量到下一条目的位置。
        // 通过 self.assemble_key 结合前一个键的共享前缀和当前条目的非共享部分，组装出完整的当前键。
        if let Some((shared, non_shared, _valsize, entry_head_len)) = self.parse_entry_and_advance() {
            self.assemble_key(current_off + entry_head_len, shared, non_shared);

            // Adjust current_restart_ix
            let num_restarts = self.number_restarts();
            // 更新当前重启点索引：
            // 检查并更新 self.current_restart_ix（当前所在的重启点索引）：当迭代位置超过下一个重启点时，将重启点索引递增，确保其始终指向当前条目所在的重启点区间。
            // 这一步是为了后续可能的 seek 操作能快速定位到正确的重启点范围。
            while self.current_restart_ix + 1 < num_restarts && self.get_restart_point(self.current_restart_ix + 1) < self.current_entry_offset {
                self.current_restart_ix += 1;
            }
            // 若解析成功，返回 true 表示迭代器已成功推进到下一个有效条目。
            true
        } else {
            // 若解析失败（如数据格式错误），在调试模式下触发 panic，否则返回 false。
            #[cfg(debug_assertions)]
            panic!("[debug mode panic] parse_entry_and_advance(): couldn't parse entry head at/after {:?}", self.key);
            #[allow(unreachable_code)]
            false
        }
    }

    /// 这段代码定义了 reset 方法，用于将 SSTable 数据块（Block）的迭代器重置到初始状态，相当于重新迭代器可以重新开始遍历数据块。
    ///
    /// 这个方法的作用是让迭代器回到初始状态，通常在以下场景使用：
    ///
    /// 数据块遍历结束后，需要重新开始遍历时（如 advance 方法检测到已到末尾时会调用 reset）。
    ///
    /// 需要放弃当前的遍历进度，从头开始处理数据块时。
    ///
    /// 通过重置这些核心状态变量，迭代器可以像刚初始化时一样，重新解析数据块中的键值对条目。
    ///
    /// 简单来说，它就像迭代器的 "重置按钮"，一键清除所有遍历状态，让迭代器回到起点，准备重新开始遍历。
    ///
    /// 重置（reset()）功能
    fn reset(&mut self) {
        println!("----------block reset----------");
        // 将迭代器当前的偏移量重置为数据块的起始位置（0），表示从数据块开头开始遍历。
        self.offset = 0;
        // 重置值（value）的偏移量为 0，清除上一个值的位置信息。
        self.val_offset = 0;
        // 将当前重启点索引重置为 0，指向第一个重启点。
        self.current_restart_ix = 0;
        // 清空当前缓存的键（key），确保没有残留的上一次遍历的键信息。
        self.key.clear();
    }

    /// 这段代码定义了 prev 方法，用于将 SSTable 数据块（Block）的迭代器向后移动到上一个键值对条目，实现了迭代器的反向遍历功能。
    ///
    /// 后退操作（prev()）
    fn prev(&mut self) -> bool {
        println!("----------block prev----------");
        // as in the original implementation -- seek to last restart point, then look for key
        // 边界检查：
        // 首先记录当前条目的偏移量（orig_offset）
        let orig_offset = self.current_entry_offset;
        // At the beginning, can't go further back
        // 如果当前已在数据块起始位置（orig_offset == 0），说明无法继续向后移动，重置迭代器并返回 false
        if orig_offset == 0 {
            self.reset();
            return false;
        }
        // 通过循环调整 current_restart_ix（当前重启点索引），找到第一个起始位置小于 orig_offset 的重启点
        // 如果当前重启点索引已为 0 仍不满足条件，则特殊处理（将偏移量设为重启点数组位置，索引设为重启点总数）
        while self.get_restart_point(self.current_restart_ix) >= orig_offset {
            // todo: double check this
            if self.current_restart_ix == 0 {
                self.offset = self.restarts_off;
                self.current_restart_ix = self.number_restarts();
                break;
            }
            self.current_restart_ix -= 1;
        }

        self.offset = self.get_restart_point(self.current_restart_ix);
        assert!(self.offset < orig_offset);

        let mut result;

        // Stop if the next entry would be the original one (self.offset always points to the start
        // of the next entry)
        // 从目标重启点开始正向遍历：
        // 将迭代器偏移量设置为找到的重启点位置
        // 循环调用 advance() 正向遍历条目，直到偏移量超过 orig_offset（即到达原始位置的前一个条目）
        loop {
            result = self.advance();
            if self.offset >= orig_offset {
                break;
            }
        }
        // 返回操作结果：
        // 返回最后一次 advance() 的结果，指示是否成功定位到上一个有效条目
        result
    }

    /// 这段代码定义了一个 seek 方法，用于在 SSTable 的数据块（Block）中快速定位到第一个键大于或等于目标键（to）的位置，是存储引擎中高效查找数据的核心逻辑之一。
    ///
    /// 查找操作（seek()）
    fn seek(&mut self, to: &[u8]) {
        println!("----------block seek----------");
        // self.reset() 将迭代器重置到初始状态，准备开始新的查找
        self.reset();
        // 定义 left 和 right 作为二分查找的左右边界，范围是数据块中所有 "重启点"（restart points）的索引
        let mut left = 0;
        // SSTable 的数据块中会定期插入 "重启点"（如每 16 个键插入一个），每个重启点存储完整的键（非重启点的键通常用前缀压缩存储）
        let mut right = if self.number_restarts() == 0 {
            0
        } else {
            self.number_restarts() - 1
        };
        // Do a binary search over the restart points.
        // 对重启点进行二分查找（left 和 right 移动），快速定位到可能包含目标键的重启点区间：
        while left < right {
            let middle = (left + right + 1) / 2;
            // 通过 seek_to_restart_point(middle) 定位到中间重启点
            self.seek_to_restart_point(middle);
            // 比较该重启点的键与目标键 to
            let c = self.opt.cmp.cmp(&self.key, to);

            if c == Ordering::Less {
                // 若重启点的键 小于 目标键，调整左边界（left = middle）
                left = middle;
            } else {
                // 若重启点的键 大于等于 目标键，调整右边界（right = middle - 1）
                right = middle - 1;
            }
        }

        assert_eq!(left, right);
        // 二分查找结束后，定位到最可能的重启点（current_restart_ix = left）
        self.current_restart_ix = left;
        self.offset = self.get_restart_point(left);

        // Linear search from here on
        // 从该重启点开始进行线性遍历（self.next()），逐个比较键，直到找到第一个 大于或等于 目标键 to 的位置并返回
        while let Some((k, _)) = self.next() {
            if self.opt.cmp.cmp(k.as_slice(), to) >= Ordering::Equal {
                return;
            }
        }
    }

    /// 这段代码定义了 valid 方法，用于判断当前 SSTable 数据块（Block）迭代器的状态是否有效（即是否指向一个可用的键值对条目）。
    ///
    /// 这个方法的作用是提供迭代器状态的有效性检查，避免在无效状态下（如迭代到数据块末尾、未正确解析条目等）尝试读取数据而导致错误。
    /// 在迭代过程中（如 next、seek 等操作后），通常会先调用 valid() 判断状态，再进行后续的数据访问。
    ///
    /// 简单来说，它是迭代器的 "状态检测器"，用于确认当前是否有可用的键值对可以读取。
    ///
    /// 有效性（valid()）
    fn valid(&self) -> bool {
        println!("----------block valid----------");
        // !self.key.is_empty()：检查当前缓存的键（self.key）是否非空。如果键为空，说明迭代器未指向任何有效条目。
        // self.val_offset > 0：验证值的偏移量（self.val_offset）是否大于 0。val_offset 记录了值在数据块中的起始位置，若为 0 通常表示未正确定位到值。
        // self.val_offset <= self.restarts_off：确保值的偏移量不超过重启点数组的起始偏移量（self.restarts_off）。因为重启点数组位于数据块的末尾，值的位置必须在重启点之前才是有效的数据区域。
        // 只有当这三个条件同时满足时，方法才返回 true，表示迭代器当前指向一个有效的键值对，可以安全地读取其键和值。
        !self.key.is_empty() && self.val_offset > 0 && self.val_offset <= self.restarts_off
    }

    /// 这段代码定义了 current 方法，用于获取当前迭代器所指向的键值对（key-value），是 SSTable 数据块迭代器中获取数据的核心接口。
    ///
    /// 获取当前键值对（current()）
    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------block current----------");
        // 首先通过 self.valid() 判断迭代器当前是否处于有效状态（即是否指向一个可用的键值对）。
        // 只有在 valid() 返回 true 时，才会继续获取键值对；否则返回 None。
        if self.valid() {
            // 将键和值封装成 Some((key, value)) 返回，供调用方使用。
            Some((
                // 获取键:
                // 通过 self.key.clone().freeze() 处理当前缓存的键：
                // self.key 是之前通过 assemble_key 组装好的完整键（以可修改的字节容器形式存在）。
                // clone() 复制该键的内容，freeze() 将其转换为不可修改的 Bytes 类型（一种高效的字节容器，适合在 Rust 中传递和共享）。
                self.key.clone().freeze(),
                // 获取值:
                // 通过 self.block.slice(self.val_offset..self.offset) 截取值的内容：
                // self.block 是整个数据块的原始字节数据。
                // self.val_offset 是值在数据块中的起始偏移量，self.offset 是值的结束偏移量（同时也是下一个条目的起始位置）。
                // slice 方法截取这个区间的字节，作为值的内容（同样以 Bytes 类型返回）。
                self.block.slice(self.val_offset..self.offset),
            ))
        } else {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::block_builder::BlockBuilder;
    use crate::options;
    use crate::test_util::{test_iterator_properties, LdbIteratorIter};
    use crate::types::{current_key_val, LdbIterator};

    fn get_data() -> Vec<(&'static [u8], &'static [u8])> {
        vec![
            (b"key1", b"value1"),
            (b"loooooooooooooooooooooooooooooooooongerkey1", b"shrtvl1"),
            ("medium length key 1".as_bytes(), "some value 2".as_bytes()),
            (b"prefix_key1", b"value"),
            (b"prefix_key2", b"value"),
            (b"prefix_key3", b"value"),
        ]
    }

    #[test]
    fn test_block_iterator_properties() {
        let o = options::for_test();
        let mut builder = BlockBuilder::new(o.clone());
        let mut data = get_data();
        // 它会将集合的长度缩减到 n（这里是 4）
        // 如果原集合长度小于等于 4，则该操作无效果
        // 如果原集合长度大于 4，则只保留前 4 个元素，后面的元素会被丢弃
        data.truncate(4);
        for &(k, v) in data.iter() {
            builder.add(k, v);
        }
        let block_contents = builder.finish();

        let block = Block::new(o.clone(), block_contents).iter();
        test_iterator_properties(block);
    }

    #[test]
    fn test_block_empty() {
        let mut o = options::for_test();
        o.block_restart_interval = 16;
        let builder = BlockBuilder::new(o);

        let blockc = builder.finish();
        assert_eq!(blockc.len(), 8);
        assert_eq!(blockc, vec![0, 0, 0, 0, 1, 0, 0, 0]);

        let block = Block::new(options::for_test(), blockc);

        LdbIteratorIter::wrap(&mut block.iter()).for_each(|_| {
            panic!("expected 0 iterations");
        });
    }

    #[test]
    fn test_block_build_iterate() {
        let data = get_data();
        let mut builder = BlockBuilder::new(options::for_test());

        for &(k, v) in data.iter() {
            builder.add(k, v);
        }

        let block_contents = builder.finish();
        let mut block = Block::new(options::for_test(), block_contents).iter();
        let mut i = 0;

        assert!(!block.valid());

        for (k, v) in LdbIteratorIter::wrap(&mut block) {
            assert_eq!(&k[..], data[i].0);
            assert_eq!(v, data[i].1);
            i += 1;
        }
        assert_eq!(i, data.len());
    }

    #[test]
    fn test_block_iterate_reverse() {
        let mut o = options::for_test();
        o.block_restart_interval = 3;
        let data = get_data();
        let mut builder = BlockBuilder::new(o.clone());

        for &(k, v) in data.iter() {
            builder.add(k, v);
        }

        let block_contents = builder.finish();
        let mut block = Block::new(o.clone(), block_contents).iter();

        assert!(!block.valid());
        assert_eq!(block.next(), Some((b"key1".to_vec(), b"value1".to_vec())));
        assert!(block.valid());
        block.next();
        assert!(block.valid());
        block.prev();
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"key1".to_vec(), b"value1".to_vec()))
        );
        block.prev();
        assert!(!block.valid());

        // Verify that prev() from the last entry goes to the prev-to-last entry
        // (essentially, that next() returning None doesn't advance anything)
        while block.next().is_some() {}

        block.prev();
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"prefix_key2".to_vec(), b"value".to_vec()))
        );
    }

    #[test]
    fn test_block_seek() {
        let mut o = options::for_test();
        o.block_restart_interval = 3;

        let data = get_data();
        let mut builder = BlockBuilder::new(o.clone());

        for &(k, v) in data.iter() {
            builder.add(k, v);
        }

        let block_contents = builder.finish();

        let mut block = Block::new(o.clone(), block_contents).iter();

        block.seek(b"prefix_key2");
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"prefix_key2".to_vec(), b"value".to_vec()))
        );

        block.seek(b"prefix_key0");
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"prefix_key1".to_vec(), b"value".to_vec()))
        );

        block.seek(b"key1");
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"key1".to_vec(), b"value1".to_vec()))
        );

        block.seek(b"prefix_key3");
        assert!(block.valid());
        assert_eq!(
            current_key_val(&block),
            Some((b"prefix_key3".to_vec(), b"value".to_vec()))
        );

        block.seek(b"prefix_key8");
        assert!(!block.valid());
        assert_eq!(current_key_val(&block), None);
    }

    #[test]
    fn test_block_seek_to_last() {
        let mut o = options::for_test();

        // Test with different number of restarts
        for block_restart_interval in [2, 6, 10] {
            o.block_restart_interval = block_restart_interval;

            let data = get_data();
            let mut builder = BlockBuilder::new(o.clone());

            for &(k, v) in data.iter() {
                builder.add(k, v);
            }

            let block_contents = builder.finish();

            let mut block = Block::new(o.clone(), block_contents).iter();

            block.seek_to_last();
            assert!(block.valid());
            assert_eq!(
                current_key_val(&block),
                Some((b"prefix_key3".to_vec(), b"value".to_vec()))
            );
        }
    }
}
